package mspbots.autotask.client.core.client;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import lombok.extern.log4j.Log4j2;
import mspbots.autotask.common.QueryBody;
import mspbots.autotask.common.RabbitKeys;
import mspbots.autotask.common.ResultSync;
import mspbots.autotask.common.annotation.ApiException;
import org.springframework.amqp.rabbit.core.RabbitMessagingTemplate;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.util.UriComponentsBuilder;
import org.springframework.web.util.UriUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Map;

/**
 * mspbots.core.autotask.core.AutoTaskClient
 *
 * @author <a href="https://github.com/vnobo">Alex bob</a>
 * @date Created by 2020/7/9
 */
@Log4j2
@Service
public class AutoTaskClient {

    private final String INTEGRATION_CODE = "FIGVTZG52JRLRJFUYQCDEJZM7B";

    private final WebClient webClient;
    private final RabbitMessagingTemplate rabbitMessagingTemplate;
    private final ObjectMapper objectMapper;

    public AutoTaskClient(WebClient.Builder builder,
                          RabbitMessagingTemplate rabbitMessagingTemplate,
                          ObjectMapper objectMapper) {
        this.webClient = builder
                .defaultHeader("Content-Type", "application/json").build();
        this.rabbitMessagingTemplate = rabbitMessagingTemplate;
        this.objectMapper = objectMapper;
    }

    public Mono<JsonNode> get(URI uri) {
        log.debug("Request get uri: [{}]", uri.toString());
        return Mono.deferContextual(ctx -> {
            ClientProperties clientProperties = ctx.get("properties");
            ResultSync resultSync = ctx.getOrDefault("resultSync", ResultSync.withDefault(uri.getRawPath()));
            assert resultSync != null;
            resultSync.setElapsedTime(System.currentTimeMillis());
            resultSync.setTenantId(clientProperties.getTenantId());
            ObjectNode objectNode = this.objectMapper.createObjectNode();
            objectNode.putPOJO("URI", uri);
            objectNode.putPOJO("queryParams", this.objectMapper
                    .convertValue(UriComponentsBuilder.fromUri(uri).build().getQueryParams(), JsonNode.class));
            resultSync.setParams(objectNode);
            return this.webClient.get().uri(uri)
                    .headers(headers -> headers.setAll(Map.of(
                            "ApiIntegrationcode", INTEGRATION_CODE,
                            "UserName", clientProperties.getUserName(),
                            "Secret", clientProperties.getSecret())))
                    .retrieve()
                    .onStatus(HttpStatus::isError, clientResponse -> clientResponse.bodyToMono(String.class)
                            .defaultIfEmpty(this.objectMapper.createObjectNode()
                                    .set("errors", this.objectMapper.createArrayNode().add("UserName or Secret is wrong")).toString())
                            .map(errNode -> this.handlerException(errNode, resultSync.setStatus(clientResponse.statusCode()))))
                    .bodyToMono(JsonNode.class)
                    .timeout(Duration.ofMinutes(5))
                    .doOnSuccess(resp -> this.handlerSuccessNext(resp, resultSync));
        });
    }

    public Flux<JsonNode> fetch(URI url, QueryBody queryBody) {
        return this.post(url, queryBody)
                .delayElement(Duration.ofMillis(200))
                .expand(jsonNode -> this.autoNextPage(jsonNode, queryBody))
                .delayElements(Duration.ofMillis(200))
                .flatMap(jsonNode -> Flux.fromIterable(jsonNode.withArray("items")));
    }

    public Mono<JsonNode> post(URI uri, Object queryBody) {
        log.debug("Request post uri: [{}], body: [{}]", uri, queryBody);
        return Mono.deferContextual(ctx -> {
            ClientProperties clientProperties = ctx.get("properties");
            ResultSync resultSync = ctx.getOrDefault("resultSync", ResultSync.withDefault(uri.getRawPath()));
            assert resultSync != null;
            resultSync.setElapsedTime(System.currentTimeMillis());
            resultSync.setTenantId(clientProperties.getTenantId());
            ObjectNode objectNode = this.objectMapper.createObjectNode();
            objectNode.putPOJO("URI", uri);
            objectNode.putPOJO("queryBody", queryBody);
            resultSync.setParams(objectNode);
            return this.webClient.post().uri(uri)
                    .headers(headers -> headers.setAll(Map.of(
                            "ApiIntegrationcode", INTEGRATION_CODE,
                            "UserName", clientProperties.getUserName(),
                            "Secret", clientProperties.getSecret())))
                    .bodyValue(queryBody)
                    .retrieve()
                    .onStatus(HttpStatus::isError, clientResponse -> clientResponse.bodyToMono(String.class)
                            .defaultIfEmpty(this.objectMapper.createObjectNode()
                                    .set("errors", this.objectMapper.createArrayNode().add("UserName or Secret is wrong")).toString())
                            .map(errNode -> this.handlerException(errNode, resultSync.setStatus(clientResponse.statusCode()))))
                    .bodyToMono(JsonNode.class)
                    .timeout(Duration.ofMinutes(5))
                    .doOnSuccess(resp -> this.handlerSuccessNext(resp, resultSync));
        });
    }

    private Mono<JsonNode> autoNextPage(JsonNode resp, QueryBody queryBody) {

        JsonNode nextPageUrlNode = resp.findValue("nextPageUrl");

        if (ObjectUtils.isEmpty(nextPageUrlNode) || nextPageUrlNode.isNull()) {
            return Mono.empty();
        }
        String httpStr = UriUtils.decode(nextPageUrlNode.textValue(),
                StandardCharsets.UTF_8);
        return this.post(UriComponentsBuilder.fromHttpUrl(httpStr).build().toUri(), queryBody);

    }

    private ApiException handlerException(Object errNode, ResultSync resultSync) {
        resultSync.setElapsedTime(System.currentTimeMillis() - resultSync.getElapsedTime());
        resultSync.setMsg(errNode);

        log.error("sync {} core error action {}, elapsedTime {}, msg: {}",
                resultSync.getTenantId(), resultSync.getAction(),
                resultSync.getElapsedTime(), resultSync.getMsg());

        this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                RabbitKeys.LOGS_TASK_QUEUE_KEY, resultSync);
        return new ApiException(resultSync.getStatus(), errNode);
    }

    private void handlerSuccessNext(JsonNode resp, ResultSync resultSync) {
        JsonNode pageDetailsNode = resp.get("pageDetails");
        if (!ObjectUtils.isEmpty(pageDetailsNode)) {
            resultSync.setCount(pageDetailsNode.get("count").intValue());
            ObjectNode objectNode = resultSync.getParams().deepCopy();
            objectNode.putPOJO("pageDetails", pageDetailsNode);
            resultSync.setParams(objectNode);
        } else {
            resultSync.setCount(resp.size());
        }
        resultSync.setElapsedTime(System.currentTimeMillis() - resultSync.getElapsedTime());
        log.debug("Sync [{}] core success, action [{}], elapsedTime [{}], count [{}]",
                resultSync.getTenantId(), resultSync.getAction(),
                resultSync.getElapsedTime(), resultSync.getCount());

        this.rabbitMessagingTemplate.convertAndSend(RabbitKeys.BASE_EXCHANGE_NAME,
                RabbitKeys.LOGS_TASK_QUEUE_KEY, resultSync);
    }
}
